package com.winit.cloudlink.message;

import com.winit.cloudlink.common.compress.Compress.CompressCodec;
import com.winit.cloudlink.common.compress.CompressFactory;
import com.winit.cloudlink.common.compress.DefaultCompressFactory;
import com.winit.cloudlink.common.exception.CompressionException;
import com.winit.cloudlink.config.ApplicationOptions;
import com.winit.cloudlink.config.Metadata;
import com.winit.cloudlink.message.exception.RejectedMessageException;
import com.winit.cloudlink.common.utils.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

import java.io.IOException;
import java.util.UUID;


/**
 * Created by stvli on 2015/11/19.
 */
public class FastJsonMessageConverter implements MessageConverter {

	private static final Logger logger = LoggerFactory.getLogger(FastJsonMessageConverter.class);

	public static final String CONTEXT_ACCEPT_ENCODING = "Accept-Encoding";

	private static CompressFactory compressFactory = new DefaultCompressFactory();

	private volatile String defaultCharset = Constants.DEFAULT_CHARSET;
	private boolean createMessageIds = false;
	private boolean compressEnabled = Constants.DEFAULT_COMPRESSION_ENABLED;
	private CompressCodec compressCodec = Constants.DEFAULT_COMPRESSION_CODEC;
	private int noncompressMaxByte = Constants.DEFAULT_NO_COMPRESSION_SIZE;

	public FastJsonMessageConverter(Metadata metadata) {
		ApplicationOptions applicationOptions = metadata.getApplicationOptions();
		compressEnabled = applicationOptions.isCompressEnabled();
		compressCodec = applicationOptions.getCompressCodec();
		noncompressMaxByte = applicationOptions.getNonCompressMaxBytes();
		initializeJsonObjectMapper();
	}

	/**
	 * Specify the default charset to use when converting to or from text-based
	 * Message body content. If not specified, the charset will be "UTF-8".
	 *
	 * @param defaultCharset The default charset.
	 */
	public void setDefaultCharset(String defaultCharset) {
		this.defaultCharset = (defaultCharset != null) ? defaultCharset : Constants.DEFAULT_CHARSET;
	}

	public String getDefaultCharset() {
		return defaultCharset;
	}

	/**
	 * Flag to indicate that new messages should have unique identifiers added
	 * to their properties before sending. Default false.
	 *
	 * @param createMessageIds the flag value to set
	 */
	public void setCreateMessageIds(boolean createMessageIds) {
		this.createMessageIds = createMessageIds;
	}

	/**
	 * Flag to indicate that new messages should have unique identifiers added
	 * to their properties before sending.
	 *
	 * @return the flag value
	 */
	protected boolean isCreateMessageIds() {
		return createMessageIds;
	}

	/**
	 * Subclass and override to customize.
	 */
	protected void initializeJsonObjectMapper() {
		//ParserConfig.getGlobalInstance().addAccept("com.winit.");
	}

	@Override
	public Object fromMessage(Message message) throws MessageConversionException {
		Object payload = null;
		MessageProperties properties = message.getMessageProperties();
		if (properties != null) {
			String contentType = properties.getContentType();
			if (contentType != null && contentType.contains("json")) {
				try {
					byte[] messageBytes = decompressBytes(message.getBody(), properties);
					logger.info("Received message : " + new String(messageBytes));
					payload = convertBytesToObject(messageBytes);
				} catch (IOException e) {
					//throw new MessageConversionException("Failed to convert Message content", e);
					logger.error("Failed to convert Message content", e);
				} catch (CompressionException e) {
					logger.error("Failed to convert Message content", e);
					//throw new MessageConversionException("Failed to extract data", e);
				} catch (Throwable e) {
					logger.error("Failed to convert Message content", e);
				}
			} else {
				logger.warn("Could not convert incoming message with content-type [" + contentType + "]");
			}
		}
		if (payload == null) {
			payload = message.getBody();
		}
		return payload;
	}

	public final Message toMessage(Object object,
								   MessageProperties messageProperties) throws MessageConversionException {
		if (messageProperties == null) {
			messageProperties = new MessageProperties();
		}
		Message message = createMessage(object, messageProperties);
		messageProperties = message.getMessageProperties();
		if (this.createMessageIds && messageProperties.getMessageId() == null) {
			messageProperties.setMessageId(UUID.randomUUID().toString());
		}
		return message;
	}

	private byte[] decompressBytes(byte[] body, MessageProperties messageProperties) throws CompressionException {
		Object acceptEncoding = messageProperties.getHeaders().get(CONTEXT_ACCEPT_ENCODING);
		if (null == acceptEncoding) {
			return body;
		}

		String compressCodec = acceptEncoding.toString();

		CompressCodec type = CompressCodec.nameOf(compressCodec);

		return compressFactory.decompress(type, body);
	}

	private Object convertBytesToObject(byte[] body) throws IOException {
		return JSON.parseObject(body);
	}

	protected Message createMessage(Object objectToConvert,
									MessageProperties messageProperties) throws MessageConversionException {

		try {
			byte[] messageBytes = objectToConvert instanceof String ? ((String) objectToConvert).getBytes(defaultCharset) : JSON.toJSONBytes(objectToConvert);
			// 是否对消息体进行压缩
			CompressCodec dataCompressCodec = null;
			if (!compressEnabled || CompressCodec.NONE.equals(compressCodec) || messageBytes.length < noncompressMaxByte) {
				dataCompressCodec = CompressCodec.NONE;
			} else {
				messageBytes = compressFactory.compress(compressCodec, messageBytes);
				dataCompressCodec = compressCodec;
			}
			messageProperties.setHeader(CONTEXT_ACCEPT_ENCODING, dataCompressCodec.name());
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
			messageProperties.setContentEncoding(getDefaultCharset());
			if (messageBytes != null) {
				messageProperties.setContentLength(messageBytes.length);
			}
			return new Message(messageBytes, messageProperties);
		} catch (IOException e) {
			throw new MessageConversionException("Failed to convert Message content", e);
		} catch (CompressionException e) {
			throw new MessageConversionException("Failed to compress data", e);
		} catch (RejectedMessageException e) {
			throw new MessageConversionException("Failed to compress data", e);
		}
	}

}
